package org.ziegler.log.analysis.core;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;

import org.apache.commons.lang3.time.StopWatch;

import org.ziegler.log.analysis.AppConfig;
import org.ziegler.log.analysis.LogCore;
import org.ziegler.log.analysis.fileFilter.FileFilter;
import org.ziegler.log.analysis.resultFilter.ResultFilter;
import org.ziegler.log.analysis.resultHandler.ResultHandler;

public abstract class BaseProcess {

    /** 文件过滤器 */
    protected final FileFilter fileFilter;

    /** 结果过滤器 */
    protected final List<ResultFilter> resultFilters = new ArrayList<>();

    /** 结果处理器 */
    protected final List<ResultHandler> resultHandlers = new ArrayList<>();

    /** 运算计时器 */
    private final StopWatch watch = StopWatch.create();

    public BaseProcess(FileFilter fileFilter) {
        this.fileFilter = Objects.requireNonNull(fileFilter);

        LogCore.debug("Process装配fileFilter = " + fileFilter);
    }

    /**
     * 结果过滤器
     * @param resultFilter
     */
    public void addResultFilter(ResultFilter resultFilter) {
        Objects.requireNonNull(resultFilter);

        resultFilters.add(resultFilter);

        LogCore.debug("Process装配resultFilter = " + resultFilter);
    }

    /**
     * 结果处理器
     * @param resultHandler
     */
    public void addResultHandler(ResultHandler resultHandler) {
        Objects.requireNonNull(resultHandler);

        resultHandlers.add(resultHandler);

        LogCore.debug("Process装配resultHandler = " + resultHandler);
    }

    /** 准备阶段 */
    protected void prepare() {
        // 初始化线程池
        final int threadPoolNum = AppConfig.getInstance().getThreadPoolNum();
        ProcessThreadPool.SetThreadPoolNum(threadPoolNum);
    }

    /**
     * 并发计算
     */
    public void run() {

        LogCore.debug("Process准备");

        prepare();

        watch.reset();
        watch.start();

        // 并发计算

        /** 等待返回的有效future数量 */
        CompletionService<List<SplitColumn>> completionService = new ExecutorCompletionService<>(ProcessThreadPool.executorService);
        int futureCount = 0;

        for (File file : fileFilter.getFiles()) {
            final Future<List<SplitColumn>> f = completionService.submit(() -> {
                // 载入文件，并返回所有解析后的数据
                return ReaderUtils.loadLog(file);
            });

            if (f != null) {
                ++futureCount;
            }
        }

        // 与上面的文件IO读取同时进行，并行处理
        for (int i = 0; i < futureCount; i++) {
            try {
                final Future<List<SplitColumn>> f = completionService.take();
                final List<SplitColumn> splitColumns = f.get();
                splitColumns.forEach(splitColumn -> {
                    // 添加结果数据
                    resultHandlers.forEach(resultHandler -> {
                        if (resultFilters.stream().allMatch(p -> p.test(splitColumn))) {
                            // 往不同的ResultHandler中添加数据
                            resultHandler.addResult(splitColumn);
                        }
                    });
                });
            } catch (ExecutionException | InterruptedException e) {
                System.out.println("e.getMessage() = " + e.getMessage());
                e.printStackTrace();
            }
        }

        // 所有日志添加处理完毕
        resultHandlers.forEach(resultHandler -> resultHandler.addComplete());

        watch.stop();

        // 结束
        finish();

        LogCore.debug("Process结束");
    }

    /**
     * 结束
     */
    protected void finish() {
        // 输出结果
        resultHandlers.forEach(ResultHandler::output);

        // 线程池关闭
        ProcessThreadPool.shutdown();
    }

    public long getRunTime() {
        return watch.getTime();
    }

}
